package com.atguigu

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._

/**
 * @ClassName StreamWordCount
 * @Description 流处理计算
 *             未测试，需要有一个机器专门发送消息，这边接收消息后马上统计 P7
 * @Author Mr Yang
 * @Date 2020/8/19 22:02
 * @Version 1.0
 */
object StreamWordCount {
  def main(args: Array[String]): Unit = {
    val params = ParameterTool.fromArgs(args)
    val host: String = params.get("host")
    val port: Int = params.getInt("port")

    //创建流处理的环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //接收socket数据流
    val textDataStream = env.socketTextStream(host, port)

    //逐一读取数据，分词之后进行wordCount
    val wordCountDataStream = textDataStream.flatMap(_.split("\\s"))
      .filter(_.nonEmpty)
      .map((_, 1))
      .keyBy(0)
      .sum(1)

    //打印输出
    wordCountDataStream.print()

    //执行任务
    env.execute("steam word count job")
  }
}
